package com.leeyu.kafka.consumergroup;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class GroupAConsumer3 {

    public static void main(String[] args) {
        // 设置属性
        Properties properties = new Properties();
        // 连接地址
        properties.put("bootstrap.servers", "localhost:9092");
        // key value反序列化
        properties.put("key.deserializer", StringDeserializer.class);
        properties.put("value.deserializer", StringDeserializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "groupA");

        // 构建Kafka消费者者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        try {
            // 订阅主题
            consumer.subscribe(Collections.singletonList("topic-g"));
            while (true) {
                try {
                    // 设置每秒从Kafka拉取消息
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));

                    records.forEach(record -> {
                        String key = record.key();
                        String value = record.value();
                        int partition = record.partition();
                        System.out.println("消费消息成功：分区：" + partition + ", key：" + key + ", value：" + value);
                    });
                } catch (Exception e) {
                    System.out.println("消费消息异常！");
                    e.printStackTrace();
                }
            }
        } finally {
            consumer.close();
        }
    }}
